//
// Created by benxb on 2021/11/3.
//

#include <type/value.h>
#include <iostream>
#include <cstdlib>
#include <cmath>
#include "compress/compressor.h"
#include "common/logger.h"
#include "common/exception.h"

namespace benxdb {

bool Compressor::DeltaEncode(char *in, size_t in_len, benxdb::TypeId type, std::string& out) {
    if (type == TypeId::BIGINT || type == TypeId::TIMESTAMP) {
        return DeltaEncodeBigInt(in, in_len, out);
    } else if (type == TypeId::DECIMAL) {
        return DeltaEncodeDecimal(in, in_len, -1, out);
    } else if (type == TypeId::INTEGER) {
        return DeltaEncodeInt(in, in_len, out);
    } else if (type== TypeId::FLOAT) {
        return DeltaEncodeFloat(in, in_len, 6, out);
    } else {
        throw NotImplementedException("DeltaEncode not implement");
    }
}

/**
 * Encoded data_ in such format:
 * --------------------------------------------------------------
 * | Deleta bit | Data count | Min value | compressed delta ....
 * --------------------------------------------------------------
 * this function encode data_ every 8 byte data_
 * so there may be compressed delta data_ span two big int
 */
bool Compressor::DeltaEncodeBigInt(char *in, size_t in_len, std::string& out) {
    int64_t max_value, min_value;
    GetMaxMinInt64(in, in_len, max_value, min_value);
    double max_delta = max_value - min_value;
    if (max_delta >= BUSTUB_INT64_MAX) {
        out.assign(in, in_len);
        return true;
    }
    size_t in_count = in_len/sizeof(int64_t);

    uint64_t delta_size = BUSTUB_INT8_MAX;
    uint8_t delta_bit = 8; // number of bit of data_ delta
    // if max delta is greater than max uint64, will not compress
    while (max_delta > delta_size) {
        delta_size = delta_size << 1;
        delta_bit++;
    }

    // padding to 64 bit data_
    int padding_size = (in_count*delta_bit+63)/64*8;
    out.resize(sizeof(uint8_t) + sizeof(int64_t) + sizeof(int64_t) + padding_size);
    out[0] = delta_bit;
    memcpy((void *) (out.data() + sizeof(uint8_t)), &in_count, sizeof(int64_t));
    memcpy((void *) (out.data() + sizeof(uint8_t) + sizeof(int64_t)), &min_value, sizeof(int64_t));

    // compose is 64 bit and may contain more than one delta data_
    // when compose is full, we put it into out data_
    uint64_t compose = 0;
    char *out_pos = const_cast<char *>(out.data() + sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint64_t));
    uint8_t left_bit = 0;
    for (int i = 0; i < in_count; i++) {
        // left shift data_ to compress
        int8_t shift_bit = 64 - delta_bit - left_bit;
        uint64_t delta = *reinterpret_cast<uint64_t*>(in + i * sizeof(int64_t)) - min_value;
        uint64_t next_compose = 0;
        // if compose data_ is not full, compress an other data_ into it
        if (shift_bit > 0) {
            delta = delta << shift_bit;
            compose |= delta;
            left_bit += delta_bit;
            if (i+1 < in_count) {
                continue;
            }
        } else {
            shift_bit = -shift_bit;
            if (shift_bit > 0) {
                next_compose = delta << (64-shift_bit);
            }
            delta = delta >> shift_bit;
            left_bit = shift_bit;
            compose |= delta;
        }
        memcpy(out_pos, &compose, sizeof(int64_t));
        compose = next_compose;
        out_pos += sizeof(int64_t);
    }

    return true;
}


/**
 * Encoded data_ in such format:
 * --------------------------------------------------------------
 * | Data count | Min value | compressed delta ....
 * --------------------------------------------------------------
 * this function encode data_ every 8 byte data_
 * so there may be compressed delta data_ span two big int
 */
bool Compressor::DeltaEncodeBigIntWithPadding(char *in, size_t in_len, std::string& out) {
    int64_t max_value, min_value;
    GetMaxMinInt64(in, in_len, max_value, min_value);
    double max_delta = max_value - min_value;
    if (max_delta >= BUSTUB_INT64_MAX) {
        out.assign(in, in_len);
        return true;
    }
    size_t in_count = in_len/sizeof(int64_t);

    out.resize(sizeof(int64_t) + sizeof(int64_t) + in_len);
    memcpy((void *) out.data(), &in_count, sizeof(int64_t));
    memcpy((void *) (out.data() + sizeof(int64_t)), &min_value, sizeof(int64_t));

    char *out_pos = const_cast<char *>(out.data() + sizeof(uint64_t) + sizeof(uint64_t));
    for (int i = 0; i < in_count; i++) {
        uint64_t delta = *reinterpret_cast<uint64_t*>(in + i * sizeof(int64_t)) - min_value;
        memcpy(out_pos, &delta, sizeof(int64_t));
        out_pos += sizeof(int64_t);
    }

    return true;
}



/**
 * Encoded data_ in such format:
 * --------------------------------------------------------------
 * | Data count | Min value | compressed delta ....
 * --------------------------------------------------------------
 * this function encode data_ every 8 byte data_
 * so there may be compressed delta data_ span two big int
 */
bool Compressor::DeltaEncodeInt32WithPadding(char *in, size_t in_len, std::string& out) {
    int32_t max_value, min_value;
    GetMaxMinInt32(in, in_len, max_value, min_value);
    double max_delta = max_value - min_value;
    if (max_delta >= BUSTUB_INT32_MAX) {
        out.assign(in, in_len);
        return true;
    }
    size_t in_count = in_len/sizeof(int32_t);

    out.resize(sizeof(int64_t) + sizeof(int64_t) + in_len);
    memcpy((void *) out.data(), &in_count, sizeof(int64_t));
    memcpy((void *) (out.data() + sizeof(int64_t)), &min_value, sizeof(int32_t));

    char *out_pos = const_cast<char *>(out.data() + sizeof(uint64_t) + sizeof(uint64_t));
    for (int i = 0; i < in_count; i++) {
        uint32_t delta = *reinterpret_cast<int32_t *>(in + i * sizeof(int32_t)) - min_value;
        memcpy(out_pos, &delta, sizeof(uint32_t));
        out_pos += sizeof(uint32_t);
    }

    return true;
}

int64_t Compressor::GetMinValue() {
    return 0;
}

void Compressor::GetMaxMinInt64(char *in, size_t len, int64_t& max_value, int64_t& min_value) {
    if (len < sizeof(min_value)) {
        LOG_DEBUG("Length is not enough.");
        return;
    }
    max_value = min_value = *reinterpret_cast<int64_t*>(in);
    for (int i = 0; i < len; i += sizeof(int64_t)) {
        int64_t cur = *reinterpret_cast<int64_t*>(in + i);
        max_value = std::max(max_value, cur);
        min_value = std::min(min_value, cur);
    }
}

void Compressor::GetMaxMinInt32(char *in, size_t len, int32_t& max_value, int32_t& min_value) {
    if (len < sizeof(min_value)) {
        LOG_DEBUG("Length is not enough.");
        return;
    }
    max_value = min_value = *reinterpret_cast<int32_t*>(in);
    for (int i = 0; i < len; i += sizeof(int32_t)) {
        int32_t cur = *reinterpret_cast<int32_t*>(in + i);
        max_value = std::max(max_value, cur);
        min_value = std::min(min_value, cur);
    }
}

/**
 * Encoded data_ in such format:
 * --------------------------------------------------------------
 * | Deleta bit | Data count | Min value | compressed delta ....
 * --------------------------------------------------------------
 * this function decode data_ every 8 byte data_
 * so there may be compressed delta data_ span two loops
 */
bool Compressor::DeltaDecodeBigInt(char *in, size_t in_len, std::string &out) {
    const int HEADER_SIZE = 17;
    if (in_len < HEADER_SIZE) {
        return false;
    }
    uint8_t delta_bit; // delta data_ bit
    uint64_t in_count; // input data_ count
    int64_t min_val; // minimal value
    memcpy(&delta_bit, in, sizeof(delta_bit));
    in += sizeof(delta_bit);
    memcpy(&in_count, in, sizeof(in_count));
    in += sizeof(in_count);
    memcpy(&min_val, in, sizeof(min_val));
    in += sizeof(min_val);

    out.resize(sizeof(uint64_t) * in_count);
    uint64_t prev_delta = 0; // delta in last loop
    uint64_t left_bit = 0; // number of bit before delta data_
    uint64_t num_bit = delta_bit; // delta_bit in current loop
    for (int i = 0; i < in_count; i++) {
        // if data_ span two 64-bit data_, calculate data_ bit in current loop
        if (left_bit + delta_bit > 64) {
            num_bit = 64-left_bit;
        }
        // number of bit after delta data_
        uint64_t right_bit = 64-left_bit-num_bit;
        uint64_t delta = 0;
        // first construct mask data_
        for (int j = 0; j < num_bit; j++) {
            delta = (delta << 1) + 1;
        }
        delta = delta << right_bit;
        // filter data_ in mask
        delta &= *reinterpret_cast<uint64_t*>(in);
        // if data_ span byte line
        // record current data_ and continue next loop
        if (left_bit + delta_bit > 64) {
            i--;
            in += 8;
            left_bit = 0;
            prev_delta = delta;
            num_bit = delta_bit - num_bit;
            continue;
        }
        // eliminate right zeros
        delta = delta >> right_bit;
        // if data_ span byte line, we need to combine them all
        if (prev_delta) {
            prev_delta = prev_delta << num_bit;
            delta |= prev_delta;
            prev_delta = 0;
        }
        uint64_t real = min_val + delta;
        memcpy(&out[i*sizeof(uint64_t)], &real, sizeof(uint64_t));
        left_bit += num_bit;
        num_bit = delta_bit;
    }
    return true;
}

/**
 * Decode decimal format ddata
 * First get the precision and calculate its multiplier
 * after decode big int data_, we need to restore raw data_
 */
bool Compressor::DeltaDecodeDecimal(char *in, size_t in_len, std::string &out) {
    int8_t precision = 0;
    memcpy(&precision, in, sizeof(precision));
    DeltaDecodeBigInt(in+sizeof(precision), in_len-sizeof(precision), out);
    int64_t multiplier = 1;
    while (precision--) {
        multiplier *= 10;
    }
    for (int i = 0; i < out.size(); i+=8) {
        const int64_t raw = *reinterpret_cast<const int64_t *>(out.data()+i);
        double data = static_cast<double>(raw)/multiplier;
        memcpy((void *) (out.data() + i), &data, sizeof(double));
    }
    return true;
}

bool Compressor::DeltaDecodeFloat(char *in, size_t in_len, std::string &out) {
    int8_t precision = 0;
    memcpy(&precision, in, sizeof(precision));
    DeltaDecodeBigInt(in+sizeof(precision), in_len-sizeof(precision), out);
    int64_t multiplier = 1;
    while (precision--) {
        multiplier *= 10;
    }
    int count = 0;
    for (int i = 0; i < out.size(); i+=8) {
        count++;
        const int64_t raw = *reinterpret_cast<const int64_t *>(out.data()+i);
        float data = static_cast<float>(raw)/multiplier;
        memcpy((void *) (out.data() + count*sizeof(float)), &data, sizeof(float));
    }
    out = out.substr(0, sizeof(float)*count);
    return true;
}

/**
 * For encode decimal data_, first change them all to big int format with precision
 * then call encode function for bit int data_
 */
bool Compressor::DeltaEncodeDecimal(char *in, size_t in_len, int8_t precision, std::string &out) {
    if (precision == -1) {
        precision = GetPrecision(in, in_len);
    }
    if (precision > 19) {
        return false;
    }
    std::string bigint_str;
    bigint_str.resize(in_len);
    int64_t multiplier = 1;
    for (int i = 0; i < precision; i++) {
        multiplier *= 10;
    }
    for (int i = 0; i < in_len; i+=8) {
        double data = *reinterpret_cast<double*>(in+i);
        int64_t bi = std::llround(data*multiplier);
        memcpy((void *) (bigint_str.data() + i), &bi, sizeof(int64_t));
    }
    std::string encode_big_int;
    DeltaEncodeBigInt(const_cast<char *>(bigint_str.data()), bigint_str.size(), encode_big_int);
    out = std::string(1, precision) + encode_big_int;
}


int8_t Compressor::GetPrecision(char *in, size_t in_len) {
    int8_t res = 0;
    for (int i = 0; i < in_len; i+=8) {
        double data = *reinterpret_cast<double *>(in);
        int8_t b = 0; // current precision
        while (data - static_cast<int>(data) != 0) {
            data *= 10;
            b++;
        }
        res = std::max(res, b);
    }
    return res;
}

bool Compressor::DeltaEncodeFloatWithPadding(char *in, size_t in_len, int8_t precision, std::string &out) {
    if (precision == -1) {
        precision = GetPrecision(in, in_len);
    }
    if (precision > 19) {
        return false;
    }
    std::string bigint_str;
    bigint_str.resize(in_len);
    int32_t multiplier = 1;
    for (int i = 0; i < precision; i++) {
        multiplier *= 10;
    }
    for (int i = 0; i < in_len; i+=sizeof(float)) {
        float data = *reinterpret_cast<float*>(in+i);
        int32_t bi = std::round(data*multiplier);
        memcpy((void *) (bigint_str.data() + i), &bi, sizeof(int32_t));
    }
    std::string encode_big_int;
    DeltaEncodeInt32WithPadding(const_cast<char *>(bigint_str.data()), bigint_str.size(), encode_big_int);
    out = std::string(1, precision) + encode_big_int;
}


bool Compressor::DeltaEncodeDecimalWithPadding(char *in, size_t in_len, int8_t precision, std::string &out) {
    if (precision == -1) {
        precision = GetPrecision(in, in_len);
    }
    if (precision > 19) {
        return false;
    }
    std::string bigint_str;
    bigint_str.resize(in_len);
    int64_t multiplier = 1;
    for (int i = 0; i < precision; i++) {
        multiplier *= 10;
    }
    for (int i = 0; i < in_len; i+=sizeof(double)) {
        double data = *reinterpret_cast<double*>(in+i);
        int64_t bi = std::round(data*multiplier);
        memcpy((void *) (bigint_str.data() + i), &bi, sizeof(int64_t));
    }
    std::string encode_big_int;
    DeltaEncodeBigIntWithPadding(const_cast<char *>(bigint_str.data()), bigint_str.size(), encode_big_int);
    out = std::string(1, precision) + encode_big_int;
}

/**
 * For encode flaot data, first change them all to big int format with precision
 * then call encode function for bit int data
 */
bool Compressor::DeltaEncodeFloat(char *in, size_t in_len, int8_t precision, std::string &out) {
    if (precision == -1) {
        precision = GetPrecision(in, in_len);
    }
    if (precision > 12) {
        return false;
    }
    std::string bigint_str;
    bigint_str.resize(in_len*2);
    int64_t multiplier = 1;
    for (int i = 0; i < precision; i++) {
        multiplier *= 10;
    }
    for (int i = 0; i < in_len; i+=sizeof(float)) {
        float data = *reinterpret_cast<float*>(in+i);
        int64_t bi = std::llround(data*multiplier);
        memcpy((void *) (bigint_str.data() + i*2), &bi, sizeof(int64_t));
    }
    std::string encode_big_int;
    DeltaEncodeBigInt(const_cast<char *>(bigint_str.data()), bigint_str.size(), encode_big_int);
    out = std::string(1, precision) + encode_big_int;
}


bool Compressor::DeltaEncodeInt(char *in, size_t in_len, std::string& out) {
    int32_t max_value, min_value;
    GetMaxMinInt32(in, in_len, max_value, min_value);
    double max_delta = max_value - min_value;
    if (max_delta >= BUSTUB_INT32_MAX) {
        out.assign(in, in_len);
        return true;
    }
    size_t in_count = in_len/sizeof(int32_t);

    uint64_t delta_size = BUSTUB_INT8_MAX;
    uint8_t delta_bit = 8; // number of bit of data_ delta
    // if max delta is greater than max uint64, will not compress
    while (max_delta > delta_size) {
        delta_size = delta_size << 1;
        delta_bit++;
    }

    // padding to 64 bit data_
    int padding_size = (in_count*delta_bit+63)/64*8;
    out.resize(sizeof(delta_bit) + sizeof(in_count) + sizeof(min_value) + padding_size);
    out[0] = delta_bit;
    memcpy((void *) (out.data() + sizeof(uint8_t)), &in_count, sizeof(int64_t));
    memcpy((void *) (out.data() + sizeof(uint8_t) + sizeof(int64_t)), &min_value, sizeof(min_value));

    // compose is 64 bit and may contain more than one delta data_
    // when compose is full, we put it into out data_
    uint64_t compose = 0;
    char *out_pos = const_cast<char *>(out.data() + sizeof(delta_bit) + sizeof(in_count) + sizeof(min_value));
    uint8_t left_bit = 0;
    for (int i = 0; i < in_count; i++) {
        // left shift data_ to compress
        int8_t shift_bit = 64 - delta_bit - left_bit;
        uint64_t delta = *reinterpret_cast<uint32_t*>(in + i * sizeof(int32_t)) - min_value;
        uint64_t next_compose = 0;
        // if compose data_ is not full, compress an other data_ into it
        if (shift_bit > 0) {
            delta = delta << shift_bit;
            compose |= delta;
            left_bit += delta_bit;
            if (i+1 < in_count) {
                continue;
            }
        } else {
            shift_bit = -shift_bit;
            if (shift_bit > 0) {
                next_compose = delta << (64-shift_bit);
            }
            delta = delta >> shift_bit;
            left_bit = shift_bit;
            compose |= delta;
        }
        memcpy(out_pos, &compose, sizeof(int64_t));
        compose = next_compose;
        out_pos += sizeof(int64_t);
    }

    return true;
}

bool Compressor::DeltaDecodeInt(char *in, size_t in_len, std::string &out) {
    const int HEADER_SIZE = 13;
    if (in_len < HEADER_SIZE) {
        return false;
    }
    uint8_t delta_bit; // delta data_ bit
    uint64_t in_count; // input data_ count
    int32_t min_val; // minimal value
    memcpy(&delta_bit, in, sizeof(delta_bit));
    in += sizeof(delta_bit);
    memcpy(&in_count, in, sizeof(in_count));
    in += sizeof(in_count);
    memcpy(&min_val, in, sizeof(min_val));
    in += sizeof(min_val);

    out.resize(sizeof(uint32_t) * in_count);
    uint64_t prev_delta = 0; // delta in last loop
    uint64_t left_bit = 0; // number of bit before delta data_
    uint64_t num_bit = delta_bit; // delta_bit in current loop
    for (int i = 0; i < in_count; i++) {
        // if data_ span two 64-bit data_, calculate data_ bit in current loop
        if (left_bit + delta_bit > 64) {
            num_bit = 64-left_bit;
        }
        // number of bit after delta data_
        uint64_t right_bit = 64-left_bit-num_bit;
        uint64_t delta = 0;
        // first construct mask data_
        for (int j = 0; j < num_bit; j++) {
            delta = (delta << 1) + 1;
        }
        delta = delta << right_bit;
        // filter data_ in mask
        delta &= *reinterpret_cast<uint64_t*>(in);
        // if data_ span byte line
        // record current data_ and continue next loop
        if (left_bit + delta_bit > 64) {
            i--;
            in += 8;
            left_bit = 0;
            prev_delta = delta;
            num_bit = delta_bit - num_bit;
            continue;
        }
        // eliminate right zeros
        delta = delta >> right_bit;
        // if data_ span byte line, we need to combine them all
        if (prev_delta) {
            prev_delta = prev_delta << num_bit;
            delta |= prev_delta;
            prev_delta = 0;
        }
        int32_t real = min_val + delta;
        memcpy(&out[i*sizeof(int32_t)], &real, sizeof(int32_t));
        left_bit += num_bit;
        num_bit = delta_bit;
    }
    return true;
}

bool Compressor::DeltaDecode(char *in, size_t in_len, TypeId type, std::string &out) {
    if (type == TypeId::BIGINT || type == TypeId::TIMESTAMP) {
        return DeltaDecodeBigInt(in, in_len, out);
    } else if (type == TypeId::DECIMAL) {
        return DeltaDecodeDecimal(in, in_len, out);
    } else if (type == TypeId::INTEGER) {
        return DeltaDecodeInt(in, in_len, out);
    } else if (type== TypeId::FLOAT) {
        return DeltaDecodeFloat(in, in_len, out);
    } else {
        throw NotImplementedException("DeltaEncode not implement");
    }
}

}